package com.butterfly.hotkey.client.core.service;

import cn.hutool.core.util.StrUtil;
import com.butterfly.hotkey.client.core.HotkeyStore;
import com.butterfly.hotkey.client.core.Refreshable;
import com.butterfly.hotkey.client.core.config.AppClientProperties;
import com.butterfly.hotkey.client.core.netty.AppNettyClient;
import com.butterfly.hotkey.client.core.netty.AppNettyClientHandler;
import com.butterfly.hotkey.client.core.netty.NettyServerModel;
import com.butterfly.hotkey.core.constant.Constants;
import io.netty.channel.Channel;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.*;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

/**
 * @author butterfly
 * @version 1.0
 */
public class WorkerService implements Refreshable {

    private static final Logger log = LoggerFactory.getLogger(Constants.CLIENT);

    private final List<NettyServerModel> WORKER_SERVERS = new CopyOnWriteArrayList<>();

    private final AppNettyClient nettyClient;

    private final AppClientProperties clientProperties;

    public WorkerService(AppClientProperties clientProperties,
                         HotkeyStore hotkeyStore) {
        NettyService nettyService = new NettyService(this, hotkeyStore);
        this.clientProperties = clientProperties;

        // 构建处理器
        AppNettyClientHandler nettyClientHandler = new AppNettyClientHandler(clientProperties, nettyService);

        // 开启client netty
        this.nettyClient = new AppNettyClient(nettyClientHandler);

        // 连接worker地址
        refreshWorkers(clientProperties.getWorkerUrls());

        // 开启各种定时器
        startSchedule();
    }

    @Override
    public void refresh() {
        refreshWorkers(clientProperties.getWorkerUrls());
    }

    /**
     * 定期重连不健康的worker的定时器
     */
    public void startSchedule() {
        Long reConnectWorkerInterval = clientProperties.getReConnectWorkerIntervalMillis();
        ScheduledExecutorService scheduledExecutorService = Executors.newSingleThreadScheduledExecutor();
        scheduledExecutorService.scheduleAtFixedRate(this::reConnectWorkers, 30, reConnectWorkerInterval, TimeUnit.MILLISECONDS);
    }

    /**
     * 选择key对应的Channel,根据hash进行负载
     *
     * @param key key
     * @return Channel
     */
    public Channel chooseChannel(String key) {
        if (StrUtil.isEmpty(key) || WORKER_SERVERS.isEmpty()) {
            return null;
        }
        int index = Math.abs(key.hashCode() % WORKER_SERVERS.size());
        return WORKER_SERVERS.get(index).getChannel();
    }

    /**
     * 增加一个新的worker
     */
    public synchronized void addWorker(String address, Channel channel) {
        boolean exist = false;
        for (NettyServerModel workerServer : WORKER_SERVERS) {
            if (!Objects.equals(workerServer.getAddress(), address)) {
                continue;
            }
            exist = true;
            break;
        }
        if (!exist) {
            NettyServerModel server = new NettyServerModel();
            server.setAddress(address);
            server.setChannel(channel);
            WORKER_SERVERS.add(server);
        }
    }

    /**
     * 判断某个worker是否已经连接过了
     */
    public boolean hasConnected(String address) {
        for (NettyServerModel workerServer : WORKER_SERVERS) {
            if (Objects.equals(address, workerServer.getAddress())) {
                return channelIsOk(workerServer.getChannel());
            }
        }
        return false;
    }

    /**
     * 获取worker是存在，但自己没连上的address集合，供重连
     */
    public List<String> getNonConnectedWorkers() {
        List<String> list = new ArrayList<>();
        for (NettyServerModel workerServer : WORKER_SERVERS) {
            if (!channelIsOk(workerServer.getChannel())) {
                list.add(workerServer.getAddress());
            }
        }
        return list;
    }

    /**
     * 定期重连不健康的worker
     */
    private void reConnectWorkers() {
        List<String> nonConnectedWorkers = getNonConnectedWorkers();
        if (nonConnectedWorkers.isEmpty()) {
            return;
        }
        log.info("trying to reConnect to these workers :{}", nonConnectedWorkers);
        connect(nonConnectedWorkers);
    }

    private boolean channelIsOk(Channel channel) {
        return channel != null && channel.isActive();
    }

    /**
     * 监听到worker信息变化后
     * 将新的worker信息和当前的进行合并，并且连接新的address
     * address例子：10.12.139.152:11111
     */
    public void refreshWorkers(List<String> addresss) {
        removeNoneUsed(addresss);

        //去连接那些在当前配置有里有，但是list里没有的
        List<String> needConnectWorkers = newWorkers(addresss);
        if (needConnectWorkers.isEmpty()) {
            return;
        }

        log.info("new workers will connect : {}", needConnectWorkers);
        connect(needConnectWorkers);
        Collections.sort(WORKER_SERVERS);
    }

    /**
     * 根据传过来的所有的worker地址，返回当前尚未连接的新的worker地址集合，用以创建新连接
     */
    private List<String> newWorkers(List<String> allAddresses) {
        Set<String> set = new HashSet<>(WORKER_SERVERS.size());
        for (NettyServerModel server : WORKER_SERVERS) {
            set.add(server.getAddress());
        }
        List<String> list = new ArrayList<>();
        for (String address : allAddresses) {
            if (!set.contains(address)) {
                list.add(address);
            }
        }
        return list;
    }

    /**
     * 移除那些在最新的worker地址集里没有的那些
     */
    private void removeNoneUsed(List<String> addresss) {
        for (NettyServerModel server : WORKER_SERVERS) {
            boolean exist = false;
            String newAddress = server.getAddress();
            for (String address : addresss) {
                if (address.equals(newAddress)) {
                    exist = true;
                    break;
                }
            }
            if (!exist) {
                log.info("worker removed: {}", newAddress);
                Channel channel = server.getChannel();
                if (channel != null) {
                    channel.close();
                }
                WORKER_SERVERS.remove(server);
            }
        }
    }

    /**
     * 处理某个worker的channel断线事件<p>
     * 可能会导致重复删除
     *
     * @param address address
     */
    public void dealChannelInactive(String address) {
        WORKER_SERVERS.removeIf(server -> Objects.equals(address, server.getAddress()));
    }

    /**
     * 连接所有的地址
     *
     * @param addresses addresses
     */
    private synchronized void connect(List<String> addresses) {
        for (String address : addresses) {
            if (hasConnected(address)) {
                continue;
            }
            // 将URL分割 IP:Port
            String[] addressPair = address.split(":");
            if (addressPair.length != 2) {
                continue;
            }
            try {
                String ip = addressPair[0];
                int port = Integer.parseInt(addressPair[1]);
                Channel channel = nettyClient.connect(ip, port);
                if (channel != null) {
                    addWorker(address, channel);
                }
            } catch (Exception e) {
                log.error("worker {} can not connect", address);
                addWorker(address, null);
            }
        }
    }

}
